package com.junbao.message.infrastructure.core;

import java.lang.reflect.Modifier;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.Marker;

import com.junbao.message.domain.enums.ProducerName;
import com.junbao.message.infrastructure.prop.MessageConfig;
import com.junbao.message.infrastructure.util.ClassUtil;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.WorkerPool;
import com.lmax.disruptor.dsl.ProducerType;

@SuppressWarnings("all")
public abstract class AbstractDisruptorCore<T> {
 
    private static final Logger LOGGER = LoggerFactory.getLogger(AbstractDisruptorCore.class);
 
    protected static Set<Class<?>> classes = new HashSet<>();
    /**消费者*/
    protected final static ArrayList<Class<?>> initMessageConsumers = new ArrayList<>();
    /**生产者*/
    protected final static ArrayList<Class<?>> initMessageProducers = new ArrayList<>();
    
    public static final MessageConfig config = new MessageConfig();
    
    static{
        String scanPackage = config.get("calzz.scan.package");
        classes = ClassUtil.getClasses(scanPackage);
        /** 遍历得到所有的消息消费者*/
        for (Class<?> clazz : classes) {
            if (isMessageConsumerChild(clazz)) {
                initMessageConsumers.add(clazz);
            }
        }
        /**遍历得到所有的消息生产者*/
        for (Class<?> clazz : classes) {
            if (isMessageProducerChild(clazz)) {
                initMessageProducers.add(clazz);
            }
        }
    }
    
    /**默认消费者数量*/
    public static final int DEFAULT_CONSUMER_LENGTH = 4;
    /**默认线程池队列长度**/
    private static final int DEFULT_QUEUE_SIZE = 2048;
    /**ringbuffer长度*/
    public static final int RINGBUFFER_SIZE = 2048;
    /** 消费者数量*/
    private static final int CONSUMER_COUNT = 10;
        
    protected static final Map<String, DataProducer> producers = new ConcurrentHashMap<>();
    
    protected static final Map<String, DataConsumer> comsumers = new ConcurrentHashMap<>();
    
    protected RingBuffer<RequestDto> ringBuffer;
    
    private WorkerPool<RequestDto> workerPool;
    
    private SequenceBarrier sequenceBarrier;
 
    /**消费者线程池*/
    private final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(setCoreThreadSize(),
                    setMaxThreadSize(),
                    1000*60,
                    TimeUnit.MILLISECONDS, 
                    new ArrayBlockingQueue(setQueueLength()),
                    new CounsumerThreadFactory(),
                    new ConsumerRejectedExecutionHandler());
    
    /**
     * 启动disruptor配置
     * @param producerType 生产者类型
     * @param ringbufferSize 
     * @param waitStrategy 等待策略
     * @param consumers 消费者
     * @param
     */
    public void start(ProducerType producerType,int ringbufferSize,WaitStrategy waitStrategy,DataConsumer[] consumers,Class<RequestDto> eventTypeClass){
        //创建ringbufferSize
        this.ringBuffer = RingBuffer.create(producerType,
                new DataEventFactory(eventTypeClass), 
                ringbufferSize, 
                waitStrategy);
        //设置squeceBarrier
        this.sequenceBarrier = this.ringBuffer.newBarrier();
        //设置工作线程池
        this.workerPool = new WorkerPool<>(this.ringBuffer, 
                this.sequenceBarrier, 
                new DataExceptionHandler(), 
                consumers);
        //添加sequeces
        this.ringBuffer.addGatingSequences(this.workerPool.getWorkerSequences());
        //启动工作池
        this.workerPool.start(this.threadPoolExecutor);
        //注册关系线程池钩子函数
        registerShutdownHook(this.threadPoolExecutor);
        LOGGER.info(this.getClass().getSimpleName() +" disruptor is started...");
    }
    
    /**
     * 设置线程池队列长度
     * @return
     */
    public int setQueueLength(){
        return DEFULT_QUEUE_SIZE;
    }
    
    /**
     * 设置核心线程池长度
     * @return
     */
    public int setCoreThreadSize(){
        return Runtime.getRuntime().availableProcessors();
    }
    
    /**
     * 设置线程池最大线程数
     * @return
     */
    public int setMaxThreadSize(){
        return (int) (Runtime.getRuntime().availableProcessors()*2) ;
    }
 
    /**
     * 注册线程停止钩子函数
     */
    protected void registerShutdownHook(final ThreadPoolExecutor threadPoolExecutor){
        Thread shutdownHook = new Thread() {
            @Override
            public void run() {
                LOGGER.info("触发hook,结束线程执行>>>>");
                if (threadPoolExecutor == null) {
                    return;
                }
                threadPoolExecutor.shutdown();
                try {
                    if (!threadPoolExecutor.awaitTermination(10, TimeUnit.SECONDS)) {
                        //取消正在执行的任务
                    	threadPoolExecutor.shutdownNow();
                    }
                } catch (Exception e) {
                	threadPoolExecutor.shutdownNow();
                }
            }
        };
        Runtime.getRuntime().addShutdownHook(shutdownHook);
    }
    
    public DataProducer getMessageProducer(ProducerName producerName) {
        DataProducer producer = this.producers.get(producerName.getProducerName());
        if (producer == null) {
            producer = createDataProducer(producerName.getProducerName());
            this.producers.put(producerName.getProducerName(), producer);
        }
        return this.producers.get(producerName.getProducerName());
    }
    
    /**
     * 根据clazz类型获取实现的消费者
     * @param clazzType 消费者类型
     * @return
     */
    protected static DataConsumer[] getmessageConsumers(Class clazzType,int consumerCount) {
        DataConsumer[] messageConsumers = new DataConsumer[consumerCount];
        try {
            for (int j = 0; j < consumerCount; j++) {
                messageConsumers[j] = (DataConsumer) clazzType.newInstance();
            }
        } catch (InstantiationException e) {
           LOGGER.error("实例化异常！",e);
        } catch (IllegalAccessException e) {
        	LOGGER.error("参数异常！",e);
        }
        return messageConsumers;
    }
 
    /**
     * 是否是消费者
     * @param clazz
     * @return
     */
    protected static boolean isMessageConsumerChild(Class<?> clazz){
        if (!Modifier.isAbstract(clazz.getModifiers()) &&  DataConsumer.class.isAssignableFrom(clazz)) {
            return true;        
        }else {
            Class<?> superclass = clazz.getSuperclass();
            if (superclass != null) {
                isMessageConsumerChild(superclass);
            }
            return false;
        }
    }
    
    /**
     * 是否是生产者
     * @param clazz
     * @return
     */
    protected static boolean isMessageProducerChild(Class<?> clazz){
        if (!Modifier.isAbstract(clazz.getModifiers()) &&  DataProducer.class.isAssignableFrom(clazz)) {
            return true;        
        }else {
            Class<?> superclass = clazz.getSuperclass();
            if (superclass != null) {
                isMessageConsumerChild(superclass);
            }
            return false;
        }
    }
    
    /**
     * 创建消息生产者
     * @param producerName 消息生产者名称
     * @return
     */
    public abstract DataProducer createDataProducer(String producerName);
    
    /**
     * 工作线程名称
     * @return
     */
    public abstract String setWorkThreadName();
    
    /**
     * 线程池拒绝策略具体实现
     * @param runnable 拒绝的线程
     * @param executor 异常
     */
    public abstract void rejectedExecutionHandler(Runnable runnable, ThreadPoolExecutor executor);
    
    /**
     * disruptor异常处理
     * @param throwable 异常类型
     * @param sequence 序列号
     * @param event 消息对象
     */
    public abstract void disruptorExceptionHandler(Throwable throwable, long sequence, Object event);
    
    /**
     * 获取消费者class类型
     * @return
     */
    public abstract Class setCounsumerClassType();
    
    /**
     * 线程工厂
     * @author yilin
     *
     */
    class CounsumerThreadFactory implements ThreadFactory{
 
        private final AtomicInteger threadNumber = new AtomicInteger(0);
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName(setWorkThreadName()+"-"+threadNumber.incrementAndGet());	
            thread.setDaemon(false);
            return thread;
        }
    }
    
    /**
     * 任务执行失败异常处理
     * @author yilin
     *
     */
    class ConsumerRejectedExecutionHandler implements RejectedExecutionHandler{
 
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            rejectedExecutionHandler(r, executor);
        }
    }
 
    /**
     * disruptor异常处理
     * @author yilin
     *
     */
    class DataExceptionHandler implements ExceptionHandler{
 
        @Override
        public void handleEventException(Throwable ex, long sequence, Object event) {
            disruptorExceptionHandler(ex, sequence, event);
        }
 
        @Override
        public void handleOnStartException(Throwable ex) {
        }
 
        @Override
        public void handleOnShutdownException(Throwable ex) {
        }
    }
 
}
